<?php
/**
 * Created by PhpStorm.
 * User: sxg
 * Date: 2022/8/2
 * Time: 16:38
 */

namespace Socket\Ms;

use Socket\Ms\Event\Event;
use Socket\Ms\Protocols\Mqtt;
use Socket\Ms\Protocols\WebSocket;

class TcpConnections {

    public $_socketFd;
    public $_clientIp;
    public $_server;

    public $_readBufferSize = 1024;
    public $_recvBufferSize = 1024 * 1000 * 10;     //当前连接接收缓冲区字节数大小
    public $_recvLen = 0;           //当前连接目前接受到的字节数
    public $_recvBufferFull = 0;    //当前连接是否超出接收缓冲区
    public $_recvBuffer = '';

    public $_sendLen = 0;
    public $_sendBuffer = '';
    public $_sendBufferSize = 1024 * 1000;
    public $_sendBufferFull = 0;

    public $_heartTime = 0;
    const HEART_BEAT = 10;

    const STATUS_CLOSE = 10;
    const STATUS_CONNECTED = 11;
    public $_status;

    public $protocol;

    public $subscribeTopic = '';


    public function __construct($socketFd, $clientIp, $server, $protocol) {
        $this->_socketFd = $socketFd;
        $this->_clientIp = $clientIp;
        $this->_server = $server;
        $this->_heartTime = time();
        $this->_status = self::STATUS_CONNECTED;
        $this->protocol = $protocol;
        stream_set_blocking($this->_socketFd, 0);
        stream_set_write_buffer($this->_socketFd, 0);
        stream_set_read_buffer($this->_socketFd, 0);
        Server::$_eventLoop->add($this->_socketFd, Event::READ, [$this, "recvSocket"]);
        //客户端上来之后，不要马上添加这个写事件，否则它会一直触发可写事件，导致执行大量的发送操作
        //会消耗cpu的资源
        //Server::$_eventLoop->add($this->_socketFd, Event::WRITE, [$this, "writeSocket"]);
    }

    public function isConnected() : bool
    {
        if ($this->_status == self::STATUS_CONNECTED && is_resource($this->_socketFd))
            return true;
        else
            return false;
    }

    public function recvSocket() {
        if ($this->_recvLen < $this->_recvBufferSize) {
            $data = fread($this->_socketFd, $this->_readBufferSize);
            if ($data === '' || $data === false) {
                if (feof($this->_socketFd) || !is_resource($this->_socketFd)) {
                    $this->close();
                }
            } else {
                //把接收到的数据放在接收缓冲区
                $this->_recvBuffer .= $data;
                $this->_recvLen += strlen($data);
                //发过来的data确实没问题 但是recvbuffer中的数据没有清除 导致没有走进这里
//                if ((ord($this->_recvBuffer[0]) & 0xf0) == Mqtt::PUBREC_PACK_TYPE) {
//                    echo "接收到订阅者的原生pubrec了,消息的长度是".strlen($this->_recvBuffer)."\r\n";
//                    echo "接收缓冲区的大小是".$this->_recvLen.PHP_EOL;
//                }
                $this->_server->onRecv();
            }
        } else {
            $this->_recvBufferFull++;
        }
        if ($this->_recvLen > 0) {
            $this->handleMessage();
        }
    }

    public function handleMessage() {
        /**@var server $server*/
        $server = $this->_server;
        if (is_object($this->protocol) && $this->protocol != null) {
            while ($this->protocol->Len($this->_recvBuffer)) {
                $msgLen = $this->protocol->msgLen($this->_recvBuffer);
                //截取一条消息
                $oneMsg = substr($this->_recvBuffer, 0, $msgLen);
                $this->_recvBuffer = substr($this->_recvBuffer, $msgLen);
                $this->_recvLen -= $msgLen;
                $server->onMsg();
                $message = $this->protocol->decode($oneMsg);
                $this->eventCallBack($message);
//                $server->eventCallBak("receive", [$message, $this]);
//                $this->resetHeartTime();
            }
        } else {
//            $server->eventCallBak("receive", [$this->_recvBuffer, $this]);
            $this->eventCallBack($this->_recvBuffer);
            $this->_recvBufferFull = 0;
            $this->_recvLen = 0;
            $this->_recvBuffer = '';
            $server->onMsg();
            $this->resetHeartTime();
        }
    }

    public function eventCallBack($message) {
        switch ($this->_server->_usingProtocol) {
            case 'stream':
            case 'text':
            case 'tcp':
                $this->_server->eventCallBak("receive", [$message, $this]);
                break;
            case 'http':
                $request = $this->createRequest();
                $response = new Response($this);
                if ($request->_request['method'] == 'OPTIONS') {
                    $response->sendAllowOrigin();
                } else {
                    $this->_server->eventCallBak("request", [$request, $response]);
                }
                break;
            case 'ws':
                if ($this->protocol->webSocketHandShakeStatus == WebSocket::WEBSOCKET_START_STATUS) {
                    //握手成功
                    if ($this->send()) {
                        if ($this->protocol->webSocketHandShakeStatus == WebSocket::WEBSOCKET_RUNNING_STATUS) {
                            $this->_server->eventCallBak("open", [$this]);
                        } else {
                            $this->close();
                        }
                    }
                } else if ($this->protocol->webSocketHandShakeStatus == WebSocket::WEBSOCKET_RUNNING_STATUS) {
                    if ($this->protocol->opcode == WebSocket::OPCODE_PING) {
                        $this->send();
                    }
                    if ($message) {
                        $this->_server->eventCallBak("message", [$this, $message]);
                    }
                } else {
                    $this->close();
                }
                break;
            case "mqtt":
                //连接报文
                switch ($this->protocol->control_packet_type) {
                    case $this->protocol::CONNECT_PACK_TYPE:
                        if ($this->protocol->protocolName != "MQTT") {
                            $this->close();
                            return false;
                        }
                        if ($this->protocol->protocolLevel != 4) {
                            $this->send();
                            $this->close();
                            return false;
                        }
                        //不合法的客户端标识符
                        if (!preg_match('/^[a-zA-Z0-9]{1,23}$/', $this->protocol->clientIdentifier)) {
//                            fwrite($this->_socketFd, $this->protocol->responseConnAckPack(0x02));
//                            $this->close();
//                            return false;
                        }
                        if ($this->_server->eventCallBak("open", [$this])) {
                            $this->send();
                        }
                        break;
                    case $this->protocol::PINGREQ_PACK_TYPE:
                        print_r("收到心跳请求了".time()."\n");
                        $this->send();
                        break;
                    case $this->protocol::SUBSCRIBE_PACK_TYPE:
                        $this->_server->eventCallBak("subscribe", [$this]);
                        $this->send();
                        break;
                    case $this->protocol::UNSUBSCRIBE_PACK_TYPE:
                        //UNSUBSCRIBE报文固定报头的第3,2,1,0位是保留位且必须分别设置为0,0,1,0 服务端必须认为任何其它的值都是不合法的并关闭连接
                        if ($this->protocol->unsubscribeReserve != 0x02) {
                            $this->close();
                        }
                        $this->_server->eventCallBak("unsubscribe", [$this]);
                        $this->send();
                        break;
                    case $this->protocol::PUBLISH_PACK_TYPE:
                        if ($this->protocol->publishPackQos == 0) {
                            $this->_server->eventCallBak("publish", [$this]);
                        }
                        if ($this->protocol->publishPackQos == 1) {
                            $this->_server->eventCallBak("publish", [$this]);
                            $this->send();
                        } else if ($this->protocol->publishPackQos == 2) {
                            $this->send();
                        }
                        break;
                    case $this->protocol::PUBACK_PACK_TYPE:
                        echo "接收到订阅者返回的PUBACK\n";
                        break;
                    case $this->protocol::PUBREL_PACK_TYPE:
//                        echo "接收到发布者返回的PUBREL\n";
                        $this->_server->eventCallBak("publish", [$this]);
                        $this->send();
                        break;
                    case $this->protocol::PUBREC_PACK_TYPE:
                        echo "接收到订阅者返回的PUBREC\n";
                        $this->send();
                        break;
                    case $this->protocol::PUBCOMP_PACK_TYPE:
                        echo "接收到订阅者返回的PUBCOMP\n";
                        break;
                    case $this->protocol::DISCONNECT_PACK_TYPE:
                        $this->close();
                        break;
                    default:
                        echo "不支持的数据包\n";
                        $this->close();
                }
                break;
        }
    }

    public function createRequest() : Request {
        $request = new Request();
        $request->_get = $_GET;
        $request->_post = $_POST;
        $request->_files = $_FILES;
        $request->_request = $_REQUEST;
        return $request;
    }

    public function close() {
        //epoll_ctl(3, EPOLL_CTL_DEL, 7, 0x7ffd0f3f32b0) = 0
        Server::$_eventLoop->del($this->_socketFd, Event::READ);
        Server::$_eventLoop->del($this->_socketFd, Event::WRITE);
        if (is_resource($this->_socketFd)) {
            fclose($this->_socketFd);
        }
        /**@var Server $server*/
        $server = $this->_server;
        $server->eventCallBak("close", [$this]);
        $server->removeClient($this->_socketFd);
        $this->_status = self::STATUS_CLOSE;
        $this->_socketFd = null;
        $this->_sendLen = 0;
        $this->_sendBuffer = '';
        $this->_sendBufferFull = 0;
        $this->_sendBufferSize = 0;

        $this->_recvBufferFull = 0;
        $this->_recvBuffer = '';
        $this->_recvBufferSize = 0;
        $this->_recvLen = 0;
    }

    public function checkHeartTime() {
        $nowTime = time();
        if ($nowTime - $this->_heartTime >= self::HEART_BEAT) {
            $this->_server->echoLog("心跳时间已超出:%d\n", $nowTime - $this->_heartTime);
            return true;
        }
        return false;
    }

    public function resetHeartTime() {
        $this->_heartTime = time();
    }

    public function needWrite()
    {
        return $this->_sendLen > 0;
    }

    public function send($data = '') {
        if (!is_resource($this->_socketFd) || feof($this->_socketFd)) {
            $this->close();
            return false;
        }
        $len = strlen($data);
        if ($this->_sendLen + $len < $this->_sendBufferSize) {
            if (is_object($this->protocol) && $this->protocol != null) {
                $bin = $this->protocol->encode($data);
                $this->_sendBuffer .= $bin[1];
                $this->_sendLen += $bin[0];
            } else {
                $this->_sendBuffer .= $data;
                $this->_sendLen += $len;
            }

            if ($this->_sendLen >= $this->_sendBufferSize) {
                $this->_sendBufferFull++;
            }
        }
        set_error_handler(function () {});
        $writeLen = fwrite($this->_socketFd, $this->_sendBuffer, $this->_sendLen);
        restore_error_handler();
        if ($writeLen == $this->_sendLen) {
            $this->_sendBuffer = '';
            $this->_sendLen = 0;
            $this->_sendBufferFull = 0;
            return true;
        } else if ($this->_sendLen > 0) {
            $this->_sendBuffer = substr($this->_sendBuffer, $writeLen);
            $this->_sendLen -= $writeLen;
            Server::$_eventLoop->add($this->_socketFd, Event::WRITE, [$this, "writeSocket"]);
        } else {
            if (!is_resource($this->_socketFd || feof($this->_socketFd))) {
                $this->close();
            }
        }
    }


    //重新封装一个send方法 因为有些客户端的control_packet_type不是publish类型 在decode时候会有问题 这里就直接发送就好 不走协议封装
    public function sendDataForMqttPublish($data): bool {
        if (!is_resource($this->_socketFd) || feof($this->_socketFd)) {
            $this->close();
            return false;
        }
        $len = strlen($data);
        if ($this->_sendLen + $len < $this->_sendBufferSize) {
            $this->_sendBuffer .= $data;
            $this->_sendLen += $len;
            if ($this->_sendLen >= $this->_sendBufferSize) {
                $this->_sendBufferFull++;
            }
        }
        set_error_handler(function () {});
        $writeLen = fwrite($this->_socketFd, $this->_sendBuffer, $this->_sendLen);
        restore_error_handler();
        if ($writeLen == $this->_sendLen) {
            $this->_sendBuffer = '';
            $this->_sendLen = 0;
            $this->_sendBufferFull = 0;
            return true;
        } else if ($this->_sendLen > 0) {
            $this->_sendBuffer = substr($this->_sendBuffer, $writeLen);
            $this->_sendLen -= $writeLen;
            Server::$_eventLoop->add($this->_socketFd, Event::WRITE, [$this, "writeSocket"]);
        } else {
            if (!is_resource($this->_socketFd || feof($this->_socketFd))) {
                $this->close();
            }
        }
        return false;
    }

    public function writeSocket() {
        if ($this->needWrite()) {
            set_error_handler(function () {});
            $writeLen = fwrite($this->_socketFd, $this->_sendBuffer, $this->_sendLen);
            restore_error_handler();
            if ($writeLen == $this->_sendLen) {
                $this->_sendBuffer = '';
                $this->_sendLen = 0;
                $this->_sendBufferFull = 0;
                Server::$_eventLoop->del($this->_socketFd, Event::WRITE);
                return true;
            }
            else if ($writeLen > 0) {
                $this->_sendBuffer = substr($this->_sendBuffer, $writeLen);
                $this->_sendLen -= $writeLen;
            } else {
                $this->close();
            }
        }
    }
}